import org.apache.spark.sql
import org.apache.spark.sql.{DataFrame, Dataset, Row, SparkSession}
import org.junit.Test

class SQLTest {
  @Test
  def dataset1(): Unit = {
    //创建SparkSession
    val spark = new sql.SparkSession.Builder()
      .master("local[6]")
      .appName("dataset")
      .getOrCreate()
    //导入隐式转换
    //这个spark是一个对象
    import spark.implicits._
    //演示
    val source = spark.sparkContext.parallelize(Seq(Person("张三", 10), Person("李四", 19)))
    val dataset: Dataset[Person] = source.toDS()
    //Dataset支持强类型API
    dataset.filter(item => item.age > 10).show()
    //Dataset支持弱类型API
    dataset.filter('age > 10).show()
    dataset.filter($"age" > 10).show()
    //Dataset可以直接编写sql表达式
    dataset.filter("age > 10").show()
  }

  @Test
  def dataset2(): Unit = {
    //创建SparkSession
    val spark = new sql.SparkSession.Builder()
      .master("local[6]")
      .appName("dataset")
      .getOrCreate()
    //导入隐式转换
    //这个spark是一个对象
    import spark.implicits._
    //演示
    val source = spark.sparkContext.parallelize(Seq(Person("张三", 10), Person("李四", 19)))
    val dataset: Dataset[Person] = source.toDS()

    dataset.explain(true)
    //无论dataset放置的是什么类型对象，转换成 rdd都是internalRow
    val executionRdd = dataset.queryExecution.toRdd
  }

  @Test
  def dataset3(): Unit = {
    //创建SparkSession
    val spark = new sql.SparkSession.Builder()
      .master("local[6]")
      .appName("dataset")
      .getOrCreate()
    //导入隐式转换
    //这个spark是一个对象
    import spark.implicits._
    //演示
    //这样子的RDD的类型是Person
    val dataset = spark.createDataset(Seq(Person("张三", 10), Person("李四", 19)))
    dataset.explain(true)
  }

  @Test
  def dataframe1(): Unit = {
    //创建SparkSession
    val spark = SparkSession.builder()
      .master("local[6]")
      .appName("dataframe")
      .getOrCreate()

    //创建DataFrame
    import spark.implicits._
    val dataFrame = Seq(Person("张三", 10), Person("李四", 19)).toDF()

    //DataFrame的语法
    dataFrame.where('age > 10)
      .select('name)
      .show()
  }

  @Test
  def dataframe2(): Unit = {
    //创建SparkSession
    val spark = SparkSession.builder()
      .master("local[6]")
      .appName("dataframe")
      .getOrCreate()

    //创建DataFrame
    import spark.implicits._
    val dataFrame = Seq(Person("张三", 10), Person("李四", 19)).toDF()

    //DataFrame的语法
    dataFrame.where('age > 10)
      .select('name)
      .show()
  }

  @Test
  def dataframe3(): Unit = {
    //创建SparkSession
    val spark = SparkSession.builder()
      .master("local[6]")
      .appName("dataframe")
      .getOrCreate()

    //创建DataFrame
    import spark.implicits._
    val dataFrame = Seq(Person("张三", 10), Person("李四", 19)).toDF()

    //DataFrame的语法
    dataFrame.where('age > 10)
      .select('name)
      .show()
  }

  @Test
  def Test(): Unit = {
    //创建SparkSession
    val spark = SparkSession.builder()
      .master("local[6]")
      .appName("Test")
      .getOrCreate()

    import spark.implicits._
    //读取数据集
    val sourceDF = spark.read
      .option("header", value = true)
      .csv("C:\\Users\\HR\\Desktop\\A.csv")

    //    sourceDF.show(10)
    //    sourceDF.printSchema()
    //处理
    sourceDF.select('Year, 'month, 'PM_Dongsi)
      .where('PM_Dongsi =!= "NA")
      .groupBy('Year, 'month)
      .count()
      .show()
    //直接使用SQL
    //注册为临时表
    sourceDF.createOrReplaceTempView("pm")
    val resultDF = spark.sql("select * from pm")
    spark.stop()
  }

  @Test
  def set_and_frame(): Unit = {
    val spark = SparkSession.builder()
      .master("local[6]")
      .appName("Test")
      .getOrCreate()
    import spark.implicits._
    val personlist = Seq(Person("张三", 10), Person("李四", 19))
    //DataFrame是弱类型,操作的是row对象
    val df: DataFrame = personlist.toDF()
    //Dataset是强类型，操作的是存储的对象
    val ds: Dataset[Person] = personlist.toDS()
  }

  @Test
  def row(): Unit = {
    //1、Rowrhecjian。它是什么
    val p = Person("zhangsan", 15)
    //row对象必须配合Schema对象，才会有列名
    val row = Row("zhangsan", 15)
    //2、从Row中获取数据
    row.getString(0)
    row.getInt(1)
    //row类也是样例类
    row match {
      case Row(name, age) => println(name, age)
    }

  }
}

case class Person(name: String, age: Int)